package com.lagou.cl.handler;

import com.alibaba.fastjson.JSONObject;
import com.lagou.cl.config.Constant;
import com.lagou.cl.config.RpcProperties;
import com.lagou.cl.model.RpcRequest;
import com.lagou.cl.model.RpcResponse;
import com.lagou.cl.util.Obj2ByteUtil;
import com.lagou.cl.util.SpringBeanUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author 程  林
 * @date 2020-05-23 18:40
 * @description //
 * @since V1.0.0
 */
@Slf4j
@Component
public class AccountServiceHandler extends ChannelInboundHandlerAdapter {

	private final CuratorFramework curatorFramework;

	private final SpringBeanUtil springBean;

	private final RpcProperties rpcProperties;

	public AccountServiceHandler(CuratorFramework curatorFramework, SpringBeanUtil springBean, RpcProperties rpcProperties) {
		this.curatorFramework = curatorFramework;
		this.springBean = springBean;
		this.rpcProperties = rpcProperties;
		init();
	}


	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

		RpcResponse response = new RpcResponse();
		String requestId = null;
		Object result = null;
		Object errorMsg = null;

		log.info("接受到参数：{} ", msg);


		//msg：RpcRequest.class
		//判断数据是否调用的方法
		if (Objects.isNull(msg)) {
			errorMsg = "请求参数为空";
			return;
		}

		if (msg instanceof RpcRequest) {
			RpcRequest request = (RpcRequest) msg;

			String reqId = request.getRequestId();
			if (StringUtils.isBlank(reqId)) {
				errorMsg = "请求Id为空";
				return;
			}
			requestId = reqId;

			//调用的那个服务类
			String className = request.getClassName();
			if (StringUtils.isBlank(className)) {
				errorMsg = "服务类名为空";
				return;
			}
			//调用服务类中的那个方法
			String methodName = request.getMethodName();
			if (StringUtils.isBlank(className)) {
				errorMsg = "方法名为空";
				return;
			}

			Class<?> aClass = Class.forName(className);
			Object beanByType = springBean.getBeanByType(aClass);
			if (null == beanByType) {
				errorMsg = "服务类对象不存在，请检查...";
				return;
			}

			//找到对应的方法
			Method method = aClass.getMethod(methodName, request.getParameterTypes());
			if (null == method) {
				errorMsg = "方法不存在，请检查...";
				return;
			}

			try {
				result = method.invoke(beanByType, request.getParameters());
			} catch (Exception e) {
				errorMsg = e.getMessage();
				log.info("调用目标方法，异常...");
			}
		}

		boolean isSuccess = true;
		if (Objects.nonNull(errorMsg)) {
			isSuccess = false;
			response.setErrorMsg(errorMsg);
		}
		response.setRequestId(requestId);
		response.setIsSuccess(isSuccess);
		response.setResult(result);

		//响应客户端
		ctx.writeAndFlush(JSONObject.toJSONString(response));

		//Zookeeper记录每个服务端的最后一次响应时间，有效时间为5秒，5s内如果该服务端没有新的请求，响应时间清零或失效
		lastTimeResponse();

	}


	/**
	 * Zookeeper记录每个服务端的最后一次响应时间，有效时间为5秒，5s内如果该服务端没有新的请求，响应时间清零或失效
	 */
	public void lastTimeResponse() throws Exception {
		lasted = System.currentTimeMillis();

		//当前负载均衡节点存在，则更新时间
		if (curatorFramework.checkExists().forPath(producerLoadBalanceZnodeName) != null) {
			update(lasted);
			log.info("本次响应时间：{} \r\n", lasted);
		} else {
			//不存在则创建负载均衡节点
			curatorFramework.create()
					.creatingParentsIfNeeded()
					.withMode(CreateMode.EPHEMERAL)
					.forPath(producerLoadBalanceZnodeName, Obj2ByteUtil.longToByteArray(lasted));
			exist = true;
			log.info("创建负载均衡节点：{}", producerLoadBalanceZnodeName);
		}

	}

	private void update(Long time) throws Exception {
		Stat stat = new Stat();
		curatorFramework.getData().storingStatIn(stat).forPath(producerLoadBalanceZnodeName);
		//更新当前版本的时间
		curatorFramework.setData().withVersion(stat.getVersion()).forPath(producerLoadBalanceZnodeName, Obj2ByteUtil.longToByteArray(time));
	}


	private Long lasted;
	private long timeOut = 5000L;
	private boolean exist = false;
	private String producerLoadBalanceZnodeName;

	/**
	 * 每隔0.1s去检查负载均衡的节点上的时间，如果节点上的时间已经大于5s，则删除该负载均衡的节点
	 */
	private void init() {
		producerLoadBalanceZnodeName = "/" + Constant.LOADBALANCE + "/" + rpcProperties.getProducerName();

		new ScheduledThreadPoolExecutor(1).scheduleWithFixedDelay(() -> {
			//每隔0.1s去检查负载均衡的节点上的时间，如果节点上的时间已经大于5s，则删除该负载均衡的节点
			if (exist && lasted != 0L && System.currentTimeMillis() - lasted >= timeOut) {
				log.info("上次响应时间：{} ，{} ms 没有被调用，delete node {}", lasted, timeOut, producerLoadBalanceZnodeName);
				try {
					curatorFramework.delete().forPath(producerLoadBalanceZnodeName);
					exist = false;
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}, 0, 100L, TimeUnit.MILLISECONDS);

	}


}
